package net.bw.realtime.tmall.dws.funcion1;

import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import net.bw.realtime.tmall.common.utils.JdbcUtil;
import net.bw.realtime.tmall.dwd.bean.TransactionEventBean;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author: liuyawei
 * @Date: 2025/5/27
 */

//todo a 实时交易额统计
public class ToTradeGMVWindowFunction {
	public static void toTradeGMV(DataStream<TransactionEventBean> timeStream) {
		SingleOutputStreamOperator<String> tradeGMV = timeStream
				.filter(x -> "pay_success".equals(x.getActionType()))
				.keyBy(x -> "all")
				.window(TumblingEventTimeWindows.of(Time.seconds(10)))
				.aggregate(new AggregateFunction<TransactionEventBean, Double, Double>() {
					@Override
					public Double createAccumulator() {
						return 0.0;
					}

					@Override
					public Double add(TransactionEventBean TransactionEventBean, Double aDouble) {
						Double amount = TransactionEventBean.getPrice() * TransactionEventBean.getQuantity();
//						System.out.println(TransactionEventBean.getPrice()+"------->"+TransactionEventBean.getQuantity()+"------->"+amount+"-----?"+TransactionEventBean.getTs()/60/1000);
						return aDouble + amount;
					}

					@Override
					public Double getResult(Double aDouble) {
						return aDouble;
					}

					@Override
					public Double merge(Double aDouble, Double acc1) {
						return aDouble + acc1;
					}
				}, new ProcessWindowFunction<Double, String, String, TimeWindow>() {
					private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
					@Override
					public void process(String s, ProcessWindowFunction<Double, String, String, TimeWindow>.Context context, Iterable<Double> iterable, Collector<String> collector) throws Exception {
						Double next = iterable.iterator().next();
						String window_start = sdf.format(new Date(context.window().getStart()));
						String window_end = sdf.format(new Date(context.window().getEnd()));

						collector.collect(window_start + "," + window_end + "," + next + "," + System.currentTimeMillis());
					}
				})
				.name("gmv-aggregation")
				.uid("gmv-aggregation");

		//tradeGMV.print();

		JdbcUtil.sinkToClickhouseUpsert(tradeGMV,
				"INSERT INTO jtp_tmall_report.dws_order_total_amount_window_report (\n" +
				"    window_start_time, window_end_time, total_amount, ts\n" +
				") VALUES (\n" +
				"             ?, ?, ?, ?\n" +
				"         )"
		);




	}


}
